package com.atguigu.bigdata.spark.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

public class SparkStreaming06_State_Join_JAVA {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(3));

        //输入DStream
        JavaReceiverInputDStream<String> ds1 = jssc.socketTextStream("localhost",8888);
        JavaReceiverInputDStream<String> ds2 = jssc.socketTextStream("localhost",9999);

        JavaPairDStream<String,Integer> ds1_pair = ds1.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 8);
            }
        });

        JavaPairDStream<String, Integer> ds2_pair = ds2.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 9);
            }
        });

        JavaPairDStream<String,Tuple2<Integer, Integer>> res =  ds1_pair.join(ds2_pair);

        res.print();

        // 1. 启动采集器
        jssc.start();
        // 2. 等待采集器的关闭
        jssc.awaitTermination();

    }
}
